package com.pzcnet.service;

import com.pzcnet.config.EventConfig;
import com.pzcnet.core.CallBack;
import com.pzcnet.dao.EventProcessDAO;
import com.pzcnet.dao.EventPublishDAO;
import com.pzcnet.model.*;
import com.pzcnet.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * Created by xdkxlk on 2017/9/10.
 */
@Slf4j
@Service
public class EventBus {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private Queue queue;
    @Autowired
    private EventPublishDAO eventPublishDAO;
    @Autowired
    private EventProcessDAO eventProcessDAO;

    private static final Executor publishEventExecutor = Executors.newCachedThreadPool();
    private static final Executor processEventExecutor = Executors.newCachedThreadPool();

    public void scanAndPublishEvent() {
        List<EventPublish> eventPublishList = this.eventPublishDAO.selectByStatus(EventPublish.EventPublishStatus.NEW.toString());
        CountDownLatch countDownLatch = new CountDownLatch(eventPublishList.size());

        for (EventPublish eventPublish : eventPublishList) {
            final int eventPublishId = eventPublish.getId();
            publishEventExecutor.execute(() -> {
                try {
                    this.publish(eventPublish);
                } catch (Exception e) {
                    log.error(String.format("发布事件的时候发生异常, EventPublishId[id=%d]",
                            eventPublishId), e);
                    this.repeatPublish(eventPublish);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void scanAndRepeatPublishEvent() {
        List<EventPublish> eventPublishList = this.eventPublishDAO.selectNeedRepeatEvent();
        CountDownLatch countDownLatch = new CountDownLatch(eventPublishList.size());

        for (EventPublish eventPublish : eventPublishList) {
            final int eventPublishId = eventPublish.getId();
            publishEventExecutor.execute(() -> {
                try {
                    this.publish(eventPublish);
                } catch (Exception e) {
                    log.error(String.format("发布事件的时候发生异常, EventPublishId[id=%d]",
                            eventPublishId), e);

                    if (eventPublish.getRepeat() >= 3) {
                        this.errorPublish(eventPublish);
                    } else {
                        this.repeatPublish(eventPublish);
                    }
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void scanAndProcessEvent() {
        List<EventProcess> eventProcessList = this.eventProcessDAO.selectByStatus(EventProcess.EventProcessStatus.NEW.toString());
        CountDownLatch countDownLatch = new CountDownLatch(eventProcessList.size());

        for (EventProcess eventProcess : eventProcessList) {
            final int eventPublishId = eventProcess.getPublishId();
            final int eventProcessId = eventProcess.getId();
            processEventExecutor.execute(() -> {
                try {
                    this.process(eventProcess);
                } catch (Exception e) {
                    log.error(String.format("处理事件的时候发生异常, EventPublishId[id=%d] EventProcessId[id=%d]",
                            eventPublishId, eventProcessId), e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
    }

    public boolean processCallBack(EventCallBack eventCallBack) {
        EventConfig.Describe describe = EventConfig.getEventCallBackMap().get(eventCallBack.getPublishEventType());
        if (describe == null)
            return false;

        CallBack callBack = (CallBack) describe.getBean();
        EventPublish eventPublish = this.eventPublishDAO.selectByPrimaryKey(eventCallBack.getPublishId());

        if (eventPublish == null || callBack == null) {
            return false;
        }

        if (eventCallBack.isSuccess()) {
            callBack.onSuccess(new Payload(eventPublish.getPayload()),
                    new Payload(eventCallBack.getInvokeResult()));
        } else {
            EventException eventException = JSONUtil.parseJSON(eventCallBack.getInvokeResult(), EventException.class);
            callBack.onError(new Payload(eventPublish.getPayload()), eventException);
        }

        return true;
    }

    @Transactional
    public void newPublish(EventPublish eventPublish) {
        eventPublish.setEventStatus(EventPublish.EventPublishStatus.NEW.toString());

        this.eventPublishDAO.insert(eventPublish);
    }

    @Transactional
    private void publish(EventPublish eventPublish) {
        eventPublish.setEventStatus(EventPublish.EventPublishStatus.PUBLISHING.toString());

        if (this.eventPublishDAO.updateSelectiveByPrimaryKey(eventPublish) > 0) {
            this.sendMsgToMQ(JSONUtil.getJSON(eventPublish));

            eventPublish = this.eventPublishDAO.selectByPrimaryKey(eventPublish.getId());
            eventPublish.setEventStatus(EventPublish.EventPublishStatus.PUBLISHED.toString());
            this.eventPublishDAO.updateSelectiveByPrimaryKey(eventPublish);
        }
    }

    @Transactional
    private void repeatPublish(EventPublish eventPublish) {
        eventPublish = this.eventPublishDAO.selectByPrimaryKey(eventPublish.getId());

        Integer repeat = eventPublish.getRepeat();
        if (repeat != null) {
            eventPublish.setRepeat(repeat + 1);
        } else {
            eventPublish.setRepeat(0);
        }
        eventPublish.setEventStatus(EventPublish.EventPublishStatus.REPEAT.toString());

        this.eventPublishDAO.updateSelectiveByPrimaryKey(eventPublish);
    }

    @Transactional
    private void errorPublish(EventPublish eventPublish) {
        eventPublish = this.eventPublishDAO.selectByPrimaryKey(eventPublish.getId());
        eventPublish.setRepeat(eventPublish.getRepeat() + 1);
        eventPublish.setEventStatus(EventPublish.EventPublishStatus.ERROR.toString());

        this.eventPublishDAO.updateSelectiveByPrimaryKey(eventPublish);

        EventException eventException = new EventException(
                String.format("事件发布失败，EventPublishId[id=%d]", eventPublish.getId())
        );
        this.processCallBack(EventCallBack.builder()
                .publishId(eventPublish.getId())
                .publishEventType(eventPublish.getEventType())
                .isSuccess(false)
                .invokeResult(JSONUtil.getJSON(eventException))
                .build());
    }

    @Transactional
    public boolean newProcess(EventProcess eventProcess) {
        if (!this.supportProcessEventType(eventProcess.getEventType())) {
            return false;
        }
        eventProcess.setEventStatus(EventProcess.EventProcessStatus.NEW.toString());
        this.eventProcessDAO.insert(eventProcess);

        return true;
    }

    @Transactional
    private void successProcess(EventProcess eventProcess) {
        eventProcess.setEventStatus(EventProcess.EventProcessStatus.PROCESSED.toString());
        this.eventProcessDAO.updateSelectiveByPrimaryKey(eventProcess);
    }

    @Transactional
    private void errorProcess(EventProcess eventProcess) {
        eventProcess.setEventStatus(EventProcess.EventProcessStatus.ERROR.toString());
        this.eventProcessDAO.updateSelectiveByPrimaryKey(eventProcess);
    }

    public boolean supportProcessEventType(String eventType) {
        return null != EventConfig.getEventProcessMap().get(eventType);
    }

    public boolean supportCallBackEventType(String eventType) {
        return null != EventConfig.getEventCallBackMap().get(eventType);
    }

    private void process(EventProcess eventProcess) {
        try {
            EventConfig.Describe describe = EventConfig.getEventProcessMap().get(eventProcess.getEventType());
            if (describe == null) {
                return;
            }
            Payload payload = new Payload(eventProcess.getPayload());
            int parameterLength = describe.getParameterNames().length;
            Object[] parameters = new Object[parameterLength];

            for (int i = 0; i < parameterLength; i++) {
                parameters[i] = payload.get(describe.getParameterNames()[i]);
            }

            Object returnRes = describe.getMethod().invoke(describe.getBean(), parameters);

            this.successProcess(eventProcess);
            this.invokeCallBack(eventProcess, returnRes, true, null);
        } catch (Exception e) {
            if (e instanceof InvocationTargetException) {
                InvocationTargetException targetException = (InvocationTargetException) e;
                Throwable throwable = targetException.getTargetException();
                if (throwable instanceof EventException) {
                    throwable.setStackTrace(new StackTraceElement[0]);
                    this.successProcess(eventProcess);
                    this.invokeCallBack(eventProcess, throwable, false, throwable.getMessage());
                    return;
                }
            }

            log.error(e.getMessage(), e);
            this.invokeCallBack(eventProcess, null, false, e.getMessage());
            this.errorProcess(eventProcess);
        }
    }

    private void invokeCallBack(EventProcess eventProcess, Object returnRes, boolean isSuccess, String msg) {
        if (!this.supportCallBackEventType(eventProcess.getEventType())) {
            return;
        }
        EventCallBack eventCallBack = EventCallBack.builder()
                .publishId(eventProcess.getPublishId())
                .publishEventType(eventProcess.getEventType())
                .invokeResult(JSONUtil.getJSON(returnRes))
                .isSuccess(isSuccess)
                .msg(msg)
                .build();
        eventCallBack.setEventType(EventType.CALL_BACK.toString());

        this.sendMsgToMQ(JSONUtil.getJSON(eventCallBack));
    }

    private void sendMsgToMQ(String msg) {
        log.info("Send msg to MQ: " + msg);
        this.rabbitTemplate.convertAndSend(queue.getName(), msg);
    }

}
